package mq

import "github.com/nsqio/go-nsq"

// NsqProducer base mq client
type NsqProducer struct {
	addr string
}

var (
	producer *nsq.Producer
)

// NewNsqProducer new nsqMsg implementation
func NewNsqProducer(addr string) *NsqProducer {
	return &NsqProducer{addr: addr}
}

func (b *NsqProducer) initMq() (*nsq.Producer, error) {
	var err error
	if producer != nil {
		if err = producer.Ping(); err == nil {
			return producer, nil
		}
	}
	producer, err = nsq.NewProducer(b.addr, nsq.NewConfig())
	if err != nil {
		return nil, err
	}
	err = producer.Ping()
	if err != nil {
		return nil, err
	}
	producer.SetLogger(nil, 0)
	return producer, nil
}

// MultiPublish 发布多条消息
func (b *NsqProducer) MultiPublish(topic string, msgs [][]byte) error {
	producer, err := b.initMq()
	if err != nil {
		return err
	}
	defer producer.Stop()
	step := 100
	start := 0
	num := len(msgs)
	for start < num {
		end := start + step
		if end > num {
			end = num
		}
		if err = producer.MultiPublish(topic, msgs[start:end]); err != nil {
			return err
		}
		start = end
	}
	return nil
}

// Publish 发布单条消息
func (b *NsqProducer) Publish(topic string, msg []byte) error {
	producer, err := b.initMq()
	if err != nil {
		return err
	}
	defer producer.Stop()
	return producer.Publish(topic, msg)
}
